package com.yanggu.mapreduce.topN;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.*;

/**
 * @version V1.0
 * @author: YangGu
 * @date: 2020/2/8 22:04
 * @description:
 */
public class TopNReducer extends Reducer<FlowBean, NullWritable, FlowBean, NullWritable> {

    /**
     * 需要注意的是这里不能使用TreeMap或者是TreeSet。因为当总流量相同的时候，会进行覆盖
     * 这里使用的是java提供的优先队列。(本质上是最大堆和最小堆)
     */
    private PriorityQueue<FlowBean> priorityQueue = new PriorityQueue<>(10, (Comparator.comparingInt(FlowBean::getSumFlow)));

    /**
     * 这里需要说明的是当key实现了Comparable。
     * MapReduce会调用key的compareTo方法进行比较，判断是否为同一个key。如果为同一个key(即使并不是所有的字段都相同)，会传入同一个reduce方法中
     * 如果想要获取这些key(这些key只是逻辑上相同，但其中有些字段的值并不相同)
     * 如果想要获取key的值，需要对values进行遍历。
     * 这时传入在reduce方法中的传入的key只是一个引用而已，在迭代器底层遍历的是时候对key的引用指向的对象进行了修改。因此可以获取到值
     *
     * @param key
     * @param values
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(FlowBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        //这里必须要对values进行遍历。因为可能总流量相同的FlowBean进行到同一个reduce方法中
        for (NullWritable value : values) {
            //1. 获取bean对象
            FlowBean flowBean = new FlowBean();
            //这里必须要用反射创建全新的对象进行
            try {
                BeanUtils.copyProperties(flowBean, key);
            } catch (Exception e) {
                e.printStackTrace();
            }

            //2. 添加到优先队列中
            //2.1 如果小于10个，直接添加添加即可
            if (priorityQueue.size() < 10) {
                //添加到优先队列中
                priorityQueue.add(flowBean);
            } else {
                //2.2 如果大于等于10个需要与堆顶元素进行判断
                FlowBean minSumFlow = priorityQueue.peek();
                //如果新的bean对象的总流量大于堆顶bean对象(也就是堆中最小的总流量的对象)
                //这里不考虑相等的情况
                if (flowBean.getSumFlow() > minSumFlow.getSumFlow()) {
                    //取出堆顶的元素
                    priorityQueue.poll();
                    //同时添加新的元素
                    priorityQueue.add(flowBean);
                }
            }
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        int size = priorityQueue.size();
        List<FlowBean> flowBeanList = new ArrayList<>(10);
        for (int i = 0; i < size; i++) {
            //从最小堆中取出的元素
            flowBeanList.add(priorityQueue.poll());
        }

        //由于是最小堆，取出的值是按照总流量升序的，需要进行反转
        Collections.reverse(flowBeanList);

        for (FlowBean flowBean : flowBeanList) {
            context.write(flowBean, NullWritable.get());
        }
    }
}
